package com.sand.pro.flinksql_pro.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.StatementSet;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.log4j.PropertyConfigurator;

public class Kafka2PostgresSQL {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 注册SourceTable: source_sensor
        //latest-offset  earliest-offset
        tableEnv.executeSql("create table source_sensor (id string , ts bigint, vc int) with("
                + "'connector' = 'kafka',"
                + "'topic' = 'topic_source_sensor',"
                + "'properties.bootstrap.servers' = '172.28.246.66:9092,172.28.246.67:9092,172.28.246.68:9092',"
                + "'properties.group.id' = 'sqlid',"
                + "'scan.startup.mode' = 'earliest-offset',"
                + "'format' = 'json'"
                + ")");
        // 2. 注册SinkTable: sink_sensor
        tableEnv.executeSql("create table sink_sensor(id string, ts bigint, vc int,PRIMARY KEY(id) NOT ENFORCED) with("
                + "'connector' = 'jdbc',"
                + "'url' = 'jdbc:postgresql://172.28.246.149:5432/dw',"
                + "'table-name' = 'test_flinksql_upset',"
                + "'driver' = 'org.postgresql.Driver',"
                + "'username' = 'postgres',"
                + "'password' = 'qwe123QWE!@',"
                + "  'sink.buffer-flush.max-rows' = '0',"
                + "  'sink.buffer-flush.interval' = '300ms',"
                + "'sink.max-retries' = '4'"
                + ")");
//        tableEnv.executeSql("select id,ts,vc from source_sensor").print();
        // 3. 从SourceTable 查询数据, 并写入到 SinkTable
//        tableEnv.executeSql("insert into sink_sensor select * from source_sensor").await();
        tableEnv.executeSql("INSERT INTO sink_sensor SELECT id,ts,vc from source_sensor").await();
//        String query = "insert into sink_sensor select * from source_sensor where id='sensor_1'";
//        StatementSet stmtSet = tableEnv.createStatementSet();
// only single INSERT query can be accepted by `addInsertSql` method
//        stmtSet.addInsertSql(query);
//        stmtSet.addInsertSql(
//                "INSERT INTO GlassOrders SELECT product, amount FROM Orders WHERE product LIKE '%Glass%'");
// execute all statements together
//        TableResult tableResult2 = stmtSet.execute();
// get job status through TableResult
//        System.out.println(tableResult2.getJobClient().get().getJobStatus());
    }
}
